package com.momo.rabbitsingleproducer.rabbitmq.topic;

import com.alibaba.fastjson.JSONObject;
import com.momo.rabbitsingleproducer.snowFlake.SnowFlake;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;


/**
 * @program: momo-cloud
 * @description: topic 主题----》消息发送
 * 消息路由规则：四种模式（topic、direct、fanout、header）
 * topic：根据绑定关键字通配符规则匹配、比较灵活
 * direct：默认，根据routingKey完全匹配，好处是先匹配再发送
 * fanout：不需要指定routingkey，相当于群发
 * header：不太常用，可以自定义匹配规则
 * @author: Jie Li
 * @create: 2019-07-27 14:58
 **/
@Component
@Slf4j
public class TopicSingelRabbitSender {

    /**
     * 自动注入RabbitTemplate模板类
     */
    @Autowired
    private RabbitTemplate rabbitTemplate;
    private SnowFlake snowFlake = new SnowFlake(1, 1);

    /**
     * 发送消息方法调用: 构建Message消息
     *
     * @param obj
     * @param properties
     * @throws Exception
     */
    public void send(Object obj, Map<String, Object> properties) {
        Long id = snowFlake.nextId();
        // 构建回调返回的数据
        CorrelationData correlationData = new CorrelationData(String.valueOf(id));
        String msg = JSONObject.toJSONString(obj);
        Message message = MessageBuilder.withBody(msg.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")
                .setDeliveryTag(id)
                // 将CorrelationData的id 与 Message的correlationId绑定，然后关系保存起来,然后人工处理
                .setCorrelationId(correlationData.getId())
                .setMessageId(id.toString()).build();
        rabbitTemplate.setConfirmCallback(confirmCallback);
        rabbitTemplate.setReturnCallback(returnCallback);
        log.info("发送的消息为:{}", JSONObject.toJSONString(message));
        rabbitTemplate.convertAndSend(TopicSingelRabbitConfig.EXCHANGE, TopicSingelRabbitConfig.ROUTINGKEY, message, correlationData);
    }


    /**
     * 回调函数: confirm确认
     * 如果消息没有到exchange,则confirm回调,ack=false
     * 如果消息到达exchange,则confirm回调,ack=true
     */
    final ConfirmCallback confirmCallback = new ConfirmCallback() {
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            if (!ack) {
                log.error("消息发送失败：correlationData: {},cause: {}", JSONObject.toJSONString(correlationData), cause);
            } else {
                log.info("消息发送成功：correlationData: {},ack: {}", JSONObject.toJSONString(correlationData), ack);
            }
        }
    };

    /**
     * 回调函数: return返回
     * exchange到queue成功,则不回调return
     * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
     */
    final ReturnCallback returnCallback = new ReturnCallback() {
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText,
                                    String exchange, String routingKey) {
            log.error("消息丢失: exchange: {},routeKey: {},replyCode: {},replyText: {}", exchange, routingKey, replyCode, replyText);
        }
    };


}
